/*
 * trans.c
 *
 *  Created on: 2011-08-16
 *      Author: Baron Wang
 */
#include <pthread.h>
#include <stdlib.h>
#include <ev.h>
#include <event.h>
#include <unistd.h>
#include <fcntl.h>
#include <arpa/inet.h>
#include <sys/ioctl.h>
#include <sys/queue.h>
#include <netinet/tcp.h>
#include <string.h>
#include <errno.h>
#include <stdio.h>
#include "trans.h"

#ifdef __linux
#define VERSION_SEC        __attribute__   ((section (".version")))
#define WEAK               __attribute__   ((weakref))
#else
#define VERSION_SEC
#define WEAK
#endif

/*
 * These APIs are from the tsock.c, it is only used by trans.
 */
extern int t_socket(const tsock_ep *endpoint);
extern int t_bind(int sockfd, const tsock_ep *endpoint);
extern int t_listen(int sockfd);
extern int t_connect(int sockfd, const tsock_ep *endpoint, int num , int *err);
extern int t_tcp_accept(int sockfd, tsock_ep* endpoint);
extern int t_close(int sockfd);
extern ssize_t t_send(int s, const void *buf, size_t len, int flags);
extern ssize_t t_recv(int s, void *buf, size_t len, int flags);
extern ssize_t t_sendto(int sockfd, const void *msg, unsigned int len,
    int flags, const tsock_ep *to);
extern ssize_t t_recvfm(int fd, void *buf, unsigned int len, int flags,
    tsock_ep *from);

#ifdef _SCTP
extern int t_sctp_accept(int sockfd, tsock_ep* endpoint, int* num);
extern int t_sctp_peeloff(int sockfd, int aid, tsock_ep *ep, int *num);
#endif

#define t_getsockname                   getsockname
#define t_getpeername                   getpeername
#define t_setsockopt                    setsockopt
#define t_getsockopt                    getsockopt

/*
 * typedef some structures.
 */
typedef struct _thread_info             thread_info;
typedef struct _socket_info             socket_info;
typedef struct _timer_info              timer_info;
typedef LIST_HEAD(t_head, _timer_info)  t_list;
typedef LIST_ENTRY(_timer_info)         t_entry;

/*
 * The global variables.
 */
VERSION_SEC static char verbuf[128] =   "libtrans                   \n"
                                        "Version: 0.1.1             \n"
                                        "Author:  Baron Wang        \n"
                                        "MailBox: ruo.w@hotmail.com \n";

static thread_info                      *m_info = NULL;
static int                              number_of_threads = 0;
static int                              trans_init_flag = 0;

#define TRANS_INIT_MAGIC_NUM            0xFACE
#define INET6_ADDRSTRLEN_BUF            INET6_ADDRSTRLEN + 2


enum{
    S_IDEL,                             /* Idle slot                    */
    S_TRANSIENT,                        /* check out not used yet       */
    S_CONNECTING,                       /* client status, connecting    */
    S_LISTEN,                           /* server status, listening     */
    S_LISTEN_DONE,                      /* used for listen once         */
    S_CONNECT,                          /* link status, connected       */
    S_CLOSING,                          /* waiting to be closed         */

    T_IDLE,                             /* empty or expired             */
    T_INUSE                             /* the timer is now running     */
};

struct _thread_info{
    pthread_t           my_tid;         /* pthread_self()               */
    pthread_mutex_t     mutex;          /* thread mutex                 */
    int                 index;          /* its index                    */
    struct ev_loop      *base_loop;     /* the base libev loop          */
    int                 max_sockets;    /* max num of sock slot         */
    int                 active_sock;    /* used sockets                 */
    struct _socket_info *psock_info;    /* sock (slots) info array      */
    int                 fd[2];          /* awake pipe                   */
    ev_io               p_ev;           /* the pipe IO event            */
    t_list              tlist;          /* free timer list              */
};

struct _socket_info{
    int                 fd;
    int                 state;          /* socket state                 */
    void                *ulc;           /* user level context           */
    thread_info         *thread;        /* back pointer to its thread   */
    sock_cb             cb;             /* socket callback              */
    ev_io               ev;             /* the IO event                 */
    int                 prtl;           /* a redandency flag            */
    tsock_ep            l_ep;           /* local endpoint               */
    tsock_ep            r_ep;           /* remote endpoint              */
    void                *t_llc;         /* timer info                   */
    char                *recv_buf;      /* the recv buffer              */
    int                 recv_len;       /* the length of alloc buf      */

    ulong               num_read;       /* the stat read                */
    ulong               num_send;       /* the stat send                */
    /*
     * XXX The socket mutex is not used because the potential dead lock
     * But this mtx should be avaliable when sending the messages.
     *
    pthread_mutex_t        mtx;
     */
    sock_opts           opts;           /* the socket opts              */
    sctp_opts           *sctp_opts;     /* sctp opts if have            */
};

struct _timer_info{
    int                 id;
    timer_cb            on_timer;
    void                *llc;           /* low level context            */
    void                *ulc;           /* usr level context            */
    int                 state;
    int                 type;
    ev_timer            ev;
    thread_info         *thread;        /* timer attached thread        */
    t_entry             next;           /* just for clearance use       */
};

static __inline void _reset_socket_info(socket_info *info, int firsttime){

    info->fd = -1;
    info->state = S_IDEL;
    info->ulc = NULL;
    info->thread = NULL;
    /* ignore info->ev_r */
    memset(&info->cb, 0, sizeof(sock_cb));
    info->prtl = 0;
    reset_ep(&info->l_ep);
    reset_ep(&info->r_ep);
    info->t_llc = NULL;

    if(!firsttime){
        if(info->recv_buf){
            free(info->recv_buf);
        }
    }
    info->recv_buf = NULL;
    info->recv_len = 0;

    info->num_read = 0;
    info->num_send = 0;

    memset(&info->opts, 0, sizeof(sock_opts));

    if(!firsttime){
        if(info->sctp_opts){
            free(info->sctp_opts);
        }
    }
    info->sctp_opts = NULL;
    return;
}

static __inline const char* _stat_to_str(int state){
    switch(state){
        case S_IDEL:
            return "IDEL";
        case S_TRANSIENT:
            return "TRANSIENT";
        case S_CONNECTING:
            return "CONNECTING";
        case S_LISTEN:
            return "LISTENING";
        case S_CONNECT:
            return "CONNECTED";
        case S_CLOSING:
            return "CLOSING";
        default:
            return "UNKNOW";
    }
}

static __inline void _print_slot(socket_info *info){

    char buf_l[INET6_ADDRSTRLEN_BUF];
    char buf_r[INET6_ADDRSTRLEN_BUF];

    buf_l[INET6_ADDRSTRLEN] = '\0';
    buf_r[INET6_ADDRSTRLEN] = '\0';

    if(is_ep_init(&info->l_ep)){
        ep_to_addr(&info->l_ep, buf_l, INET6_ADDRSTRLEN_BUF);
    }

    if(is_ep_init(&info->r_ep)){
        ep_to_addr(&info->r_ep, buf_r, INET6_ADDRSTRLEN_BUF);
    }

    LOG_ANYWAY(
        "    socket[%d] --- %s, %s. ulc: %p in thread %d, opt: %X \n"
        "      (%lu sent) L (%s:%d) <-----> (%s:%d) R (%lu recv)\n"
        , info->fd, prtl_to_str(info->prtl), _stat_to_str(info->state)
        , info->ulc, (int)(info->thread - m_info) , info->opts.opts
        , info->num_send, buf_l, ntohs(info->l_ep.port)
        , buf_r, ntohs(info->r_ep.port), info->num_read
    );
}

void trans_print(){
    register int i;
    register int j;

    for(i = 0; i < number_of_threads; ++i){

        LOG_ANYWAY(
            "thread[%d], active: %d in max %d, t_head(%p) \n"
            , m_info[i].my_tid
            , m_info[i].active_sock
            , m_info[i].max_sockets
            , m_info[i].tlist.lh_first
        );

        for(j = 0; j < m_info[i].max_sockets; ++j){

            if(m_info[i].psock_info[j].state == S_IDEL)
                continue;

            _print_slot(&m_info[i].psock_info[j]);
        }
        LOG_ANYWAY("\n");
    }
    LOG_ANYWAY("\n");
}

static __inline int _set_no_delay(int fd, int protocol, int val){
    int res = TRANS_UNSUPPORTED;
    socklen_t len = sizeof(val);

    switch(protocol){
        case SYS_TCP:
            res = t_setsockopt(fd, SYS_TCP, TCP_NODELAY, &val, len);
            break;
        case SYS_SCTP:
            res = t_setsockopt(fd, SYS_SCTP, SCTP_NODELAY, &val, len);
            break;
        default:
            break;
    }
    return res;
}

static __inline int _set_linger(int fd, int onoff, int timeout){

    struct linger lgr;

    lgr.l_onoff = onoff;    /* off */
    lgr.l_linger = timeout;    /* 0 means no timeout */

    if(t_setsockopt(fd, SOL_SOCKET, SO_LINGER, &lgr, sizeof(int))){
        LOG(_ERR, "fail to set the linger for socket[%d], linger on/off[%d], "
            "timeout[%d], errno[%d](%s).",
            fd, onoff, timeout, errno, strerror(errno));
        return TRANS_FAILED;
    }
    return TRANS_SUECCESS;
}

static int _set_multicast(int fd, tsock_ep *local,
    tsock_ep *remote, int add_loopback){

    struct ip_mreq imr;
    struct ipv6_mreq imr6;
    int tmp;
    char buf[INET6_ADDRSTRLEN_BUF];
    buf[INET6_ADDRSTRLEN] = '\0';

    memset(&imr, 0, sizeof(struct ip_mreq));
    memset(&imr6, 0, sizeof(struct ipv6_mreq));

    /* judge type by remote peer, not local    */
    switch(remote->type){
        case AF_INET:
            imr.imr_multiaddr.s_addr = remote->addr.addr_4.s_addr;
            if(AF_INET == local->type){
                imr.imr_interface.s_addr = local->addr.addr_4.s_addr;
            }else{
                imr.imr_interface.s_addr = INADDR_ANY;
            }
            if (t_setsockopt(fd, IPPROTO_IP, IP_ADD_MEMBERSHIP, &imr,
                sizeof(struct ip_mreq))){
                LOG(_CRI, "fail to join the multicast group, "
                    "for remote IP type(IPv4), protocol(%s), addr(%s), "
                    "local IP type(%s), protocol(%s), addr(%s), socket[%d], "
                    "errno[%d](%s).",
                    prtl_to_str(remote->prtl),
                    inet_ntoa(remote->addr.addr_4),
                    type_to_str(local->type),
                    prtl_to_str(local->prtl),
                    inet_ntoa(local->addr.addr_4), fd,
                    errno, strerror(errno));
                return TRANS_FAILED;
            }

            tmp = add_loopback;
            if (t_setsockopt(fd, IPPROTO_IP, IP_MULTICAST_LOOP, &tmp,
                sizeof(int))){
                LOG(_ERR, "fail to set the loopback for mcast, "
                    "for remote IP type(IPv4), protocol(%s), addr(%s), "
                    "local IP type(%s), protocol(%s), addr(%s), socket[%d], "
                    "errno[%d](%s).",
                    prtl_to_str(remote->prtl),
                    inet_ntoa(remote->addr.addr_4),
                    type_to_str(local->type),
                    prtl_to_str(local->prtl),
                    inet_ntoa(local->addr.addr_4), fd,
                    errno, strerror(errno));

                /* ignore this error */
            }

            if (t_setsockopt(fd, IPPROTO_IP, IP_MULTICAST_IF, &imr,
                sizeof(struct ip_mreq))){
                LOG(_CRI, "fail to set the multicast IF for remote "
                    "IP type(IPv4), protocol(%s), addr(%s), local IP type(%s) "
                    "protocol(%s), addr(%s), socket[%d], errno[%d](%s).",
                    prtl_to_str(remote->prtl),
                    inet_ntoa(remote->addr.addr_4),
                    type_to_str(local->type),
                    prtl_to_str(local->prtl),
                    inet_ntoa(local->addr.addr_4),
                    fd, errno, strerror(errno));
                return TRANS_FAILED;
            }

            tmp = 255;
            if(t_setsockopt(fd, IPPROTO_IP, IP_MULTICAST_TTL, &tmp,
                sizeof(int))){
                LOG(_ERR, "fail to set the TTL 255 for mcast for remote IP "
                    "type(IPv4), protocol(%s), addr(%s), local IP type(%s), "
                    "protocol(%s), addr(%s), socket[%d], errno[%d](%s).",
                    prtl_to_str(remote->prtl),
                    inet_ntoa(remote->addr.addr_4),
                    type_to_str(local->type),
                    prtl_to_str(local->prtl),
                    inet_ntoa(local->addr.addr_4),
                    fd, errno, strerror(errno));
                /* ignore this error */
            }


            break;
        case AF_INET6:
            memcpy(&imr6.ipv6mr_multiaddr, &remote->addr.addr_6,
                sizeof(struct in6_addr));

            /* choose any address not the local address    */
            imr6.ipv6mr_interface = 0;

            if (t_setsockopt(fd, IPPROTO_IPV6, IPV6_JOIN_GROUP, &imr6,
                sizeof(struct ipv6_mreq))){

                LOG(_CRI, "fail to join the multicast group, for remote IP "
                    "type(IPv6), protocol(%s), addr(%s),"
                    "socket[%d], errno[%d](%s).",
                    prtl_to_str(remote->prtl),
                    inet_ntop(AF_INET6, &remote->addr.addr_6, buf,
                        INET6_ADDRSTRLEN),
                    fd, errno, strerror(errno));
                return TRANS_FAILED;
            }

            tmp = add_loopback;
            if (t_setsockopt(fd, IPPROTO_IPV6, IPV6_MULTICAST_LOOP, &tmp,
                sizeof(int))){
                LOG(_ERR, "fail to set the loopback for mcast for remote IP"
                    "type(IPv6), protocol(%s), addr(%s),"
                    "socket[%d], errno[%d](%s).",
                    prtl_to_str(remote->prtl),
                    inet_ntop(AF_INET6, &remote->addr.addr_6, buf,
                        INET6_ADDRSTRLEN),
                    fd, errno, strerror(errno));
                /* ignore this error */
            }

            if (t_setsockopt(fd, IPPROTO_IPV6, IPV6_MULTICAST_IF, &imr6,
                            sizeof(struct ipv6_mreq))){
                LOG(_CRI, "fail to set the multicast IF for remote IP"
                    "type(IPv6), protocol(%s), addr(%s),"
                    "socket[%d], errno[%d] (%s).",
                    prtl_to_str(remote->prtl),
                    inet_ntop(AF_INET6, &remote->addr.addr_6, buf,
                        INET6_ADDRSTRLEN),
                    fd, errno, strerror(errno));
                return TRANS_FAILED;
            }

            tmp = 255;
            if(t_setsockopt(fd, IPPROTO_IPV6, IPV6_MULTICAST_HOPS, &tmp,
                sizeof(int))){
                LOG(_ERR, "fail to set the TTL 255 for mcast for remote IP "
                    "type(IPv6), protocol(%s), addr(%s),"
                    "socket[%d], errno[%d] (%s).",
                    prtl_to_str(remote->prtl),
                    inet_ntop(AF_INET6, &remote->addr.addr_6, buf,
                        INET6_ADDRSTRLEN),
                    fd, errno, strerror(errno));
                /* ignore this error */
            }
            break;
        default:
            return TRANS_UNSUPPORTED;
            break;
    }

    return TRANS_SUECCESS;
}

static int _unset_multicast(int fd, tsock_ep *l, tsock_ep *r){

    struct ip_mreq imr;
    struct ipv6_mreq imr6;
    char buf[INET6_ADDRSTRLEN_BUF];
    buf[INET6_ADDRSTRLEN] = '\0';

    memset(&imr, 0, sizeof(struct ip_mreq));
    memset(&imr6, 0, sizeof(struct ipv6_mreq));

    /* judge type by remote peer, not local */
    switch(r->type){
        case AF_INET:
            imr.imr_multiaddr.s_addr = r->addr.addr_4.s_addr;
            if(AF_INET == l->type){
                imr.imr_interface.s_addr = l->addr.addr_4.s_addr;
            }else{
                imr.imr_interface.s_addr = INADDR_ANY;
            }
            if (t_setsockopt(fd, IPPROTO_IP, IP_DROP_MEMBERSHIP, &imr,
                sizeof(struct ip_mreq))){
                LOG(_CRI, "fail to drop the multicast group for remote IP "
                    "type(IPv4), protocol(%s), addr(%s), local IP "
                    "type(%s), protocol(%s), addr(%s), socket[%d], "
                    "errno[%d](%s).",
                    prtl_to_str(r->prtl),
                    inet_ntoa(r->addr.addr_4),
                    type_to_str(l->type),
                    prtl_to_str(l->prtl),
                    inet_ntoa(l->addr.addr_4),
                    fd, errno, strerror(errno));
                return TRANS_FAILED;
            }

            break;
        case AF_INET6:
            memcpy(&imr6.ipv6mr_multiaddr, &r->addr.addr_6,
                sizeof(struct in6_addr));

            /* choose any address not the local address    */
            imr6.ipv6mr_interface = 0;

            if (t_setsockopt(fd, IPPROTO_IPV6, IPV6_LEAVE_GROUP, &imr6,
                sizeof(struct ipv6_mreq))){

                LOG(_CRI, "fail to leave the multicast group for remote IP "
                    "type(IPv6), protocol(%s), addr(%s), socket[%d], "
                    "errno[%d](%s).",
                    prtl_to_str(r->prtl),
                    inet_ntop(AF_INET6, &r->addr.addr_6, buf,
                        INET6_ADDRSTRLEN),
                    fd, errno, strerror(errno));
                return TRANS_FAILED;
            }
            break;
        default:
            return TRANS_UNSUPPORTED;
            break;
    }

    return TRANS_SUECCESS;
}


static __inline void _ev_lock_acquire(struct ev_loop *loop){
    thread_info *u = ev_userdata (loop);
    pthread_mutex_lock(&u->mutex);
    return;
}

static __inline void _ev_lock_release(struct ev_loop *loop){
    thread_info *u = ev_userdata (loop);
    pthread_mutex_unlock(&u->mutex);
    return;
}

/*
 * a load balance thread pool checkout.
 * MAY guard by mutex but not a MUST since it vaguely calculate one thread
 */
static __inline thread_info* _chkout_thread_pool(){

    register int min_sock;
    int i;

    int j = 0;

    min_sock = m_info[0].active_sock;

    for(i = 1; i < number_of_threads; ++i){
        if(m_info[i].active_sock < min_sock){
            min_sock = m_info[i].active_sock;
            j = i;
        }
    }

    if(min_sock < m_info[0].max_sockets){
        return &(m_info[j]);
    }
    return NULL;
}

/* MUST protect by mutex */
static __inline int _chkout_socket(thread_info *trd){

    int i;

    if(trd->active_sock >= trd->max_sockets){
        LOG(_CRI, "exceed the maxium socket limit[%d]"
            "current active socket[%d], in thread[%d].",
            trd->max_sockets, trd->active_sock,
            (int)(trd - m_info));
        return TRANS_EXCEED_LIMITS;
    }

    for(i = 0; i < trd->max_sockets; ++i){
        if(S_IDEL == trd->psock_info[i].state){

            /*    will try to alloc recv_buf    */
            trd->psock_info[i].recv_buf = malloc(RECV_BUFFER_SIZE);
            if(NULL == trd->psock_info[i].recv_buf){
                LOG(_CRI, "failed to alloc recv buffer.");
                return TRANS_NOT_ENOUGH_MEM;
            }
            trd->psock_info[i].recv_len = RECV_BUFFER_SIZE;
            trd->psock_info[i].state = S_TRANSIENT;
            return i;
        }
    }
    return TRANS_NOT_FOUND;
}

static int _set_sctpopt(int fd, sctp_opts *sopts){

    /* TODO SCTP
     * a lot more than we could image
     */
#ifdef _SCTP
    int res;
    socklen_t len = sizeof(struct sctp_initmsg);;
    struct sctp_initmsg init_msg;

    if(getsockopt(fd, IPPROTO_SCTP, SCTP_INITMSG, &init_msg, &len)){
        LOG(_CRI, "fail to get sockopt for sock fd[%d]"
            "errno[%d](%s).", fd, errno, strerror(errno));
        return TRANS_FAILED;
    }

    init_msg.sinit_num_ostreams = sopts.num_output_streams;
    init_msg.sinit_max_instreams = sopts.max_input_streams;

    if(t_setsockopt(fd, IPPROTO_SCTP, SCTP_INITMSG, &init_msg, &len)){
        LOG(_CRI, "fail to set sockopt for sock fd[%d]"
            "errno[%d](%s).", fd, errno, strerror(errno));
        return TRANS_FAILED;
    }

#else
    LOG(_WRN, "don't support the sctp so no operation"
        "to t_setsockopt for socket[%d].", fd);
    /* to peace to continue    */
    return TRANS_SUECCESS;
#endif
}

/*
 * This function saves the local and remote address to the structure
 * ONLY when the connection is there.
 *
 * TODO SCTP
 * need more work
 */
static int _save_conn_addr(socket_info *info){

    struct sockaddr_storage addr_s;
    socklen_t sockaddr_len = sizeof(struct sockaddr_storage);

    /* it is supposed that sock info is enough    */
    if(SYS_SCTP == info->prtl){
        /* TODO SCTP
         * need to do it
         */

    }else{

        if(t_getsockname(info->fd, (struct sockaddr*)&addr_s, &sockaddr_len)){
            LOG(_CRI, "getsockname fail, may listen on port 0 for socket"
            "[%d], errno[%d](%s).", info->fd, errno, strerror(errno));
            return TRANS_FAILED;
        }
        /* Update the local port */
        if(sockaddr_to_ep((struct sockaddr*)&addr_s, info->prtl, &info->l_ep)){
            LOG(_CRI, "fail to update local addr for socket[%d].", info->fd);
            return TRANS_FAILED;
        }

        if(t_getpeername(info->fd, (struct sockaddr*)&addr_s, &sockaddr_len)){
            LOG(_CRI, "getpeername fail, for socket[%d], errno[%d](%s).",
                info->fd, errno, strerror(errno));
            return TRANS_FAILED;
        }
        /* Update the local port */
        if(sockaddr_to_ep((struct sockaddr*)&addr_s, info->prtl,
            &info->r_ep)){
            LOG(_CRI, "fail to update remote addr for socket[%d].", info->fd);
            return TRANS_FAILED;
        }
    }

    return TRANS_SUECCESS;
}

static __inline int _set_sctpsock_opt(int fd){
#ifdef _SCTP

    /*
     * Please NOTE,
     * This structure is very different between linux type,
     * MontaVista Linux, or even different in REHL 5 and REHL 6
     *
     *
     */
    struct sctp_event_subscribe event = {
        .sctp_data_io_event = 1,
        .sctp_association_event = 1,
        .sctp_address_event = 1,
        .sctp_send_failure_event = 1,
        .sctp_peer_error_event = 1,
        .sctp_shutdown_event = 1,
        .sctp_partial_delivery_event = 0,
#ifdef _REHL5
        .sctp_adaption_layer_event = 0,
#endif
#ifdef _REHL6
        .sctp_adaptation_layer_event = 0,
        .sctp_authentication_event = 0,
#endif
    };

    if(t_setsockopt(fd, IPPROTO_SCTP, SCTP_EVENTS, &event, sizeof(event))){
        LOG(_CRI, "subscribe sctp event fail for socket[%d], errno[%d](%s).",
            fd, errno, strerror(errno));

        return TRANS_FAILED;
    }

    return TRANS_SUECCESS;

#else
    LOG(_WRN, "don't support the sctp so no operation"
        "to t_setsockopt for socket[%d]", fd);
    /* to peace to continue */
    return TRANS_SUECCESS;
#endif
}

/* the thread main loop function */
static void * __thread_loop(void* args);

/*
 * the main timer entry and dispatcher function
 * Please note that, the entry function is used to protect the thread safe.
 * And the dispatcher, which has a callback, should be UNLOCKed first before
 * calling user callback. Or else, if user locks the threads, dead ends.
 */
static void __timer_cb_entry(struct ev_loop *loop, ev_timer *w, int revents);
static void __timer_callback(struct ev_loop *loop, ev_timer *w, int revents);
/*
 *  the main socket entry and dispatcher function, the reason why need entry
 *  is the same as the timer callback.
 */
static void __sock_cb_entry(struct ev_loop *loop, ev_io *w, int revents);
static void __sock_callback(struct ev_loop *loop, ev_io *w, int revents);

static void __pipe_callback (struct ev_loop *loop, ev_io *w, int revents);
static int __create_pipe(thread_info *pinfo);
static void __close_pipe(thread_info *pinfo);
static int __wakeup_thread(thread_info *pinfo);
static void __nanosleep(ulong msec);

static int __set_timer(socket_info *info, void** outllc, void *ulc,
    int id, int mill_sec, timer_cb cb, int type);
static int __kill_timer(timer_info *timer_llc, int id);

static int __set_sock_timer(socket_info *info, int id, timer_cb cb, int tout);
static int __kill_sock_timer(socket_info *info);
static int __socket_timer_cb(int id, void *llc, void *ulc);

static int __trans_send(socket_info *info, char* buf, int len, tsock_ep* ep);
static int __read_socket(socket_info *sinfo, int *len, tsock_ep *ep);
static int __trans_close(socket_info *info);

int trans_init(trans_opt opt){

    int res = TRANS_FAILED;

    register int i;        /* thread iterator */
    register int j;        /* socket per thread iterator */

    if(!IS_LOG_INIT){
        if(INIT_LOGGER(opt.l_opt)){
            return TRANS_FAILED;
        }
    }

    if(TRANS_INIT_MAGIC_NUM == trans_init_flag){
        LOG(_WRN, "trans thread has already there.");
        return TRANS_ALREADY_INITED;
    }

    if(0 == opt.num_of_worker){
        LOG(_CRI, "there's 0 worker thread in config.");
        return TRANS_FAILED;
    }

    /* first, create the pthread_info array, and init them    */
    m_info = malloc(sizeof(thread_info) * opt.num_of_worker);
    if(NULL == m_info){
        LOG(_CRI, "fail to create the thread pool.");
        return TRANS_NOT_ENOUGH_MEM;
    }

    /* set the thread attr */
    if(NULL != opt.t_attr){
        pthread_attr_init(opt.t_attr);
    }

    number_of_threads = opt.num_of_worker;

    for(i = 0; i < opt.num_of_worker; ++i){

        res = pthread_mutex_init(&m_info[i].mutex, NULL);
        if(0 != res){
            LOG(_CRI, "fail to create the mutex in thread[%d].", i);
            goto clean;
        }

        /* init the libev */
        m_info[i].base_loop = ev_loop_new(EVFLAG_AUTO);

        /* for some data use, it is supposed to used for thread safe loop */
        ev_set_userdata (m_info[i].base_loop, (void *)&m_info[i]);

        /*
         * These settings is for the sharing loop, but in our case,
         * there's no need.
         *
        ev_set_loop_release_cb(m_info[i].base_loop, _ev_lock_release,
            _ev_lock_acquire);
         */

        if(NULL == m_info[i].base_loop){
            LOG(_CRI, "fail to create the event loop in thread[%d].", i);
            goto clean;
        }

        /* init other variables */
        m_info[i].psock_info = NULL;
        m_info[i].max_sockets = opt.max_socket_num;
        m_info[i].active_sock = 0;
        m_info[i].index = i;
        /* init timer head */
        LIST_INIT(&m_info[i].tlist);

        /*
         * create the pipe for each worker thread.
         * this is used to awake the thread in a graceful way
         */
        res = __create_pipe(&m_info[i]);
        if(0 != res){
            LOG(_CRI, "fail to create the pipe in thread[%d].", i);
            goto clean;
        }

        m_info[i].psock_info = malloc(m_info[i].max_sockets *
            sizeof(socket_info));

        if(NULL == m_info[i].psock_info){
            LOG(_CRI, "fail to create the socket slots in thread[%d].", i);
            goto clean;
        }

        for(j = 0; j < m_info[i].max_sockets; ++j){
            _reset_socket_info(&(m_info[i].psock_info[j]), 1);
        }

        /* init the thread pool */
        res = pthread_create(&m_info[i].my_tid, opt.t_attr,
                        __thread_loop, (void*)&m_info[i]);
        if(0 != res){
            LOG(_CRI, "failed to create the thread [%d], error [%d](%s)",
                i, errno, strerror(errno));
            goto error;
        }
    }

clean:

    /* set the thread attr */
    if(NULL != opt.t_attr){
        pthread_attr_destroy(opt.t_attr);
    }

    if(0 != res){
        /* some error happend */
        goto error;
    }

    /* init sueccessfully */
    trans_init_flag = TRANS_INIT_MAGIC_NUM;
    return TRANS_SUECCESS;

error:
    if(m_info){
        for(i = 0; i < opt.num_of_worker; ++i){
            /* free sockets */
            if(m_info[i].psock_info){

                if(m_info[i].psock_info->sctp_opts){
                    free(m_info[i].psock_info->sctp_opts);
                }

                free(m_info[i].psock_info);
            }

            /* No timer entry to free */
            /* free event loop */
            ev_loop_destroy(m_info[i].base_loop);

            /* destory mutex */
            pthread_mutex_destroy(&m_info[i].mutex);
        }
        free(m_info);
        m_info = NULL;
    }
    trans_init_flag = 0;
    number_of_threads = 0;

    return TRANS_FAILED;
}

void trans_fini(void){

    register int i;
    register int j;
    register int res;

    timer_info *timer;
    void *join_args = NULL;

    LOG(_CHK, "trans_fini is called, will closed.");

    if(m_info){

        for(i = 0; i < number_of_threads; ++i){

            /* stop the timers first */
            ev_suspend(m_info[i].base_loop);

            /* free sockets    */
            if(m_info[i].psock_info){

                for(j = 0; j < m_info[i].max_sockets; ++j){

                    if(m_info[i].psock_info[j].recv_buf){
                        free(m_info[i].psock_info[j].recv_buf);
                    }

                    if(m_info[i].psock_info[j].sctp_opts){
                        free(m_info[i].psock_info[j].sctp_opts);
                    }

                    if(m_info[i].psock_info[j].t_llc){
                        __kill_sock_timer(&m_info[i].psock_info[j]);
                    }

                    if(-1 != m_info[i].psock_info[j].fd){
                        ev_io_stop(m_info[i].base_loop,
                            &m_info[i].psock_info[j].ev);
                    }

                }

                LOG(_RTM, "socket slot is clear, delete for thread[%d].", i);

                free(m_info[i].psock_info);
            }

            LOG(_RTM, "Stop the pipe for the thread[%d].", i);
            ev_io_stop(m_info[i].base_loop, &m_info[i].p_ev);
            ev_break(m_info[i].base_loop, EVBREAK_ALL);

            /* free timer entry */
            while(!LIST_EMPTY(&m_info[i].tlist)){
                timer = LIST_FIRST(&m_info[i].tlist);
                LIST_REMOVE(timer, next);
                free(timer);
            }

            __close_pipe(&m_info[i]);

            /* free event loop */
            ev_loop_destroy(m_info[i].base_loop);

            res = pthread_join(m_info[i].my_tid, &join_args);

            /* destory mutex */
            pthread_mutex_destroy(&m_info[i].mutex);


            if(join_args){
                free(join_args);
            }

            if(0 != res){
                LOG(_CRI, "fail to join the thread [%d], error [%d](%s).",
                    i, errno, strerror(errno));
                continue;
            }

            LOG(_RTM, "join thread %d.", m_info[i].index);
        }

        free(m_info);
        m_info = NULL;
    }

    FINI_LOGGER;
    trans_init_flag = 0;
    number_of_threads = 0;
    (void)_stat_to_str(0);    /* to avoid ugly warning */
}

int trans_listen(void *llc, void* ulc, void **o_llc, tsock_ep *ep,
    sock_cb *cb, sock_opts *opts, sctp_opts *sopts){

    int tmp; /* general used variable, NOT register because of address      */

    int sockfd;
    int res;
    register thread_info *info;
    register socket_info *sockinfo;

    ev_io * io;    /* the stub variable to enable -fstrict-aliasing            */

    /*
     * for the SCTP alternative local addr may be invalid but not used    if so
     * put it here to emliate warnings.
     */
    tsock_ep *alter_ep = ep + 1;
    int i;                          /* the socket slot we check out         */

    char buf[INET6_ADDRSTRLEN_BUF];

    struct sockaddr_storage addr_s;
    socklen_t sockaddr_len = sizeof(struct sockaddr_storage);

    CHECKNULL(o_llc, return TRANS_A_NULL_POINTER);
    CHECKNULL(ep, return TRANS_A_NULL_POINTER);
    CHECKNULL(cb, return TRANS_A_NULL_POINTER);
    CHECKNULL(opts, return TRANS_A_NULL_POINTER);

    if(TRANS_INIT_MAGIC_NUM != trans_init_flag){
        return TRANS_NOT_INITIALIZE;
    }

    /* the init job */
    *o_llc = 0;

    if(!(opts->opts & SOCK_OPT_CLOSE_ERR) && (NULL == cb->on_sock_err)){
        LOG(_CRI, "fail to go any furthur because SOCK_OPT_CLOSE_ERR "
            "is not set but cb on_sock_err is NULL.");
        return TRANS_CROSSCHK_FAIL;
    }

    if((SYS_SCTP == ep->prtl) && (sopts == NULL)){
        /* if it is a SCTP socket, with NULL opts will fail */
        LOG(_CRI, "a SCTP endpoint with no SCTP options, will fail.");
        return TRANS_CROSSCHK_FAIL;
    }

    sockfd = t_socket(ep);
    if(sockfd < 0){
        LOG(_CRI, "create socket fail in trans_listen, return[%d].", sockfd);
        return TRANS_FAILED;
    }

    CHKLOGLVL(L_RTM){

        buf[INET6_ADDRSTRLEN] = '\0';

        LOG(_RTM, "socket[%d] ==> Step 1, Creating socket "
            "for local address(%s), port[%d], IP type(%s), protocol(%s).",
            sockfd,
            ep_to_addr(ep, buf, INET6_ADDRSTRLEN_BUF),
            ntohs(ep->port),
            type_to_str(ep->type),
            prtl_to_str(ep->prtl));
    }

    if(SYS_TCP == ep->prtl || SYS_SCTP == ep->prtl){

        if(opts->opts & SOCK_OPT_NDLAY){
            res = _set_no_delay(sockfd, ep->prtl, 1);
            if(TRANS_SUECCESS != res){
                LOG(_CRI, "fail to set the no-delay mode for "
                    "socket fd[%d], protocol(%s), opts[%X], res[%d](%s).",
                    sockfd, prtl_to_str(ep->prtl),
                    opts->opts, res, ERROR_TO_STR(res));
                t_close(sockfd);
                return TRANS_FAILED;
            }
        }
    }

    if(opts->opts & SOCK_OPT_REUSE_ADDR){
        tmp = 1;

        if(t_setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, &tmp, sizeof(int))){
            LOG(_CRI, "fail to set reuse addr to sock[%d], reason[%d](%s).",
                sockfd, errno, strerror(errno));
            t_close(sockfd);
            return TRANS_FAILED;
        }
    }

    if((SYS_UDP == ep->prtl) && (AF_INET == ep->type)){
        tmp = 1;
        if(opts->opts & SOCK_OPT_BORADCAST){

            if(t_setsockopt(sockfd, SOL_SOCKET, SO_BROADCAST, &tmp,
                sizeof(int))){
                LOG(_CRI, "fail to set broadcast for socket fd[%d], "
                    "protocol(UDP), opts[%X], err[%d](%s).",
                    sockfd, opts->opts, errno, strerror(errno));
                t_close(sockfd);
                return TRANS_FAILED;
            }
        }
    }

    if(AF_INET == ep->type){
        if(opts->opts & SOCK_OPT_TOS){
            /* that's an IP level operation */
            if(t_setsockopt(sockfd, IPPROTO_IP, IP_TOS, &opts->ip_tos,
                sizeof(int))){
                LOG(_CRI, "fail to set IP TOS[%d] for socket fd[%d], "
                    "protocol(%s), opts[%X], err[%d](%s).",
                    opts->ip_tos, sockfd,
                    prtl_to_str(ep->prtl),
                    opts->opts, errno,
                    strerror(errno));
                t_close(sockfd);
                return TRANS_FAILED;
            }
        }
    }

    if(t_bind(sockfd, ep)){
        t_close(sockfd);
        return TRANS_FAILED;
    }

    LOG(_RTM, "socket[%d] ==> Step 2, Bind sueccessfully.", sockfd);

    if((SYS_SCTP == ep->prtl) && (opts->opts & SOCK_OPT_SCTP_ALTER)){

        /* doing a check here, it may not work, but can prevent a SEGV */
        if(0 == is_ep_init(alter_ep)){
            LOG(_CRI, "the alternative endpoint is not valid.");
            t_close(sockfd);
            return TRANS_FAILED;
        }

        /*
         * try to bind an alternative endpoint for sctp, if it is correct
         * save it in the sctp_opts in sockopts (if have)
         */
        if(TRANS_SUECCESS != t_bind(sockfd, alter_ep)){
            t_close(sockfd);
            return TRANS_FAILED;
        }


        LOG(_RTM, "socket[%d] ==> Step 2 (more for SCTP), Bind the 2ed "
            "endpoint sueccessfully.", sockfd);

        /* will save it in the sctp_opt structure later */
    }
    /*
     * get the port info, if 0 is passed in, a ephemeral port will
     * be assigned and it will die quickly.
     */
    if(t_getsockname(sockfd, (struct sockaddr*)&addr_s, &sockaddr_len)){
        LOG(_CRI, "getsockname Fail, it may listen on port 0 for "
            "socket[%d], errno[%d](%s).",
            sockfd, errno, strerror(errno));

        t_close(sockfd);
        return TRANS_FAILED;
    }

    /* Update the local port */

    /* Please NOTE, although this is an easy way to get the port, recommand not
     * doing it again like what it does here
     */
    ep->port = ((uint16_t *)&addr_s)[1];

    tmp = SOCKET_BUF_SIZE;
    if(opts->opts & SOCK_OPT_SEDBUF){
        if(opts->sbuf_size > SOCKET_BUF_SIZE){
            /* have a mini send buf that is 128k */
            tmp = opts->sbuf_size;
        }
        if(t_setsockopt(sockfd, SOL_SOCKET, SO_SNDBUF, &tmp,
            sizeof(int))){

            LOG(_WRN, "fail to set the send buffer for "
                "socket[%d], errno[%d](%s).",
                sockfd, errno, strerror(errno));
            /* continue */
        }
    }

    if(opts->opts & SOCK_OPT_RCVBUF){
        if(opts->rbuf_size > SOCKET_BUF_SIZE){
            /* have a mini send buf that is 128k */
            tmp = opts->rbuf_size;
        }

        if(t_setsockopt(sockfd, SOL_SOCKET, SO_RCVBUF, &tmp,
            sizeof(int))){
            LOG(_WRN, "fail to set the recv buffer for "
                "socket [%d], errno[%d](%s).",
                sockfd, errno, strerror(errno));
            /* continue    */
        }
    }

    if(SYS_SCTP == ep->prtl || SYS_TCP == ep->prtl){
        /*
         * turn off the linger, and should be turn on/off in
         * connect or accept based on opts
         */

        if(opts->opts & SOCK_OPT_LINGER){
            (void)_set_linger(sockfd, 1, opts->linger_time);
        }else{
            (void)_set_linger(sockfd, 0, 0);
        }

        tmp = 1;
        if(t_setsockopt(sockfd, SOL_SOCKET, SO_KEEPALIVE, &tmp, sizeof(int))){

            LOG(_ERR, "fail to turn off the keep-alive for "
                "socket [%d], errno[%d](%s).",
                sockfd, errno, strerror(errno));
            /* continue */
        }
    }

    if(opts->opts & SOCK_OPT_NON_BLOCKING){
        tmp = 1;
        if(ioctl(sockfd, FIONBIO, &tmp)){
            LOG(_CRI, "fail to ioctl non-blocking socket[%d], errno[%d](%s).",
                sockfd, errno, strerror(errno));
            t_close(sockfd);
            return TRANS_FAILED;
        }
    }

    if((SYS_TCP == ep->prtl) || (SYS_SCTP == ep->prtl)){
        if(t_listen(sockfd)){
            LOG(_CRI, "fail to listen on socket[%d], errno[%d](%s).",
                sockfd, errno, strerror(errno));
            t_close(sockfd);
            return TRANS_FAILED;
        }

        if(SYS_TCP == ep->prtl){
            (void)_set_no_delay(sockfd, SYS_TCP, 1);
        }
    }

    LOG(_RTM, "socket[%d] ==> Step 3, Listen sueccessfully.", sockfd);

    if((SYS_SCTP == ep->prtl) || (cb->on_sctp_event)){
        if(TRANS_SUECCESS != _set_sctpsock_opt(sockfd)){
            t_close(sockfd);
            return TRANS_FAILED;
        }
    }

    if(NULL == llc){
        info = _chkout_thread_pool();
        /* apparently, info can not be NULL */

    }else{
        info = ((socket_info *)llc)->thread;
    }

    if(NULL == info){
        LOG(_CRI, "no available thread pool.");
        t_close(sockfd);
        return TRANS_NOT_FOUND;
    }

    pthread_mutex_lock(&info->mutex);

    i = _chkout_socket(info);
    if(i < 0){
        t_close(sockfd);
        pthread_mutex_unlock(&info->mutex);
        LOG(_CRI, "no available socket slot for thread[%d], reason %s.",
            (int)(info - m_info), trans_strerror(i));
        return TRANS_NOT_FOUND;
    }
    info->active_sock++;

    /* get a free socket slot, save it and set it */
    sockinfo = &info->psock_info[i];

    /* Does -O2 will enroll memcpy? guess not. but memcpy is not a bad idea */
    /* memcpy(&sockinfo->opts, opts, sizeof(sock_opts)); */
    sockinfo->opts.ip_tos = opts->ip_tos;
    sockinfo->opts.linger_time = opts->linger_time;
    sockinfo->opts.opts = opts->opts;
    sockinfo->opts.rbuf_size = opts->rbuf_size;
    sockinfo->opts.sbuf_size = opts->sbuf_size;
    sockinfo->thread = info;
    sockinfo->ulc = ulc;
    sockinfo->prtl = ep->prtl;

    /*
     * if it is a SCTP, we need to
     *         1, init the sock_info sctp opt
     *         2, save the alternative local address (if have),
     */
    if(SYS_SCTP == ep->prtl){
        sockinfo->sctp_opts = malloc(sizeof(sctp_opts));
        if(NULL == sockinfo->sctp_opts){
            info->active_sock--;
            _reset_socket_info(sockinfo, 0);

            t_close(sockfd);
            pthread_mutex_unlock(&info->mutex);
            LOG(_CRI, "fail to alloc SCTP opts for socket slot.");
        }

        /* sopts is definately not NULL because checked at the beginning */
        memcpy(sockinfo->sctp_opts, sopts, sizeof(sctp_opts));
        /* reset the raddr2 because it is still listen */
        reset_ep(&sockinfo->sctp_opts->raddr2);

        if(opts->opts & SOCK_OPT_SCTP_ALTER){
            memcpy(&sockinfo->sctp_opts->laddr2, alter_ep, sizeof(tsock_ep));
        }else{
            reset_ep(&sockinfo->sctp_opts->laddr2);
        }

        if(TRANS_SUECCESS != _set_sctpopt(sockfd, sopts)){

            info->active_sock--;
            _reset_socket_info(sockinfo, 0);
            t_close(sockfd);

            pthread_mutex_unlock(&info->mutex);
            LOG(_CRI, "fail to set the sctp opts.");
            return TRANS_FAILED;
        }
    }
    memcpy(&sockinfo->cb, cb, sizeof(sock_cb));
    sockinfo->fd = sockfd;
    memcpy(&sockinfo->l_ep, ep, sizeof(tsock_ep));

    if(SYS_UDP == ep->prtl){
        sockinfo->state = S_CONNECT;
    }else{
        sockinfo->state = S_LISTEN;
    }

    *o_llc = (void*) sockinfo;

    /* add for the libev */
    io = &sockinfo->ev;
    ev_init(io, __sock_cb_entry);

    sockinfo->ev.data = (void*)sockinfo;
    /*
     * For almost all Unix/Linux IP/TCP stack, it don't suppot the transition
     * from the listen to send SYN to the established state, let it just read
     */
    ev_io_set(&sockinfo->ev, sockinfo->fd, EV_READ);
    ev_io_start(info->base_loop, &sockinfo->ev);

    pthread_mutex_unlock(&info->mutex);

    LOG(_RTM, "socket[%d] ==> Step 4 (DONE), Event added "
            "sueccessfully, for thread[%d], slot[%d]."
            , sockfd, (int)(info - m_info), i);

    __wakeup_thread(info);
    return TRANS_SUECCESS;
}

int trans_connect(void *llc, void *ulc, void **o_llc,
    tsock_ep *l_ep, tsock_ep *r_ep, sock_cb *cb,
    sock_opts *opts, sctp_opts *sopts, int timeout){

    int tmp;    /* general used variable, NOT a register because of address */
    int err;
    int conn_ret;           /* the connect() ret                            */

    int sockfd;
    int need_clean = 0;     /* an indicator if need a timer to clean        */
    register thread_info *info;
    register socket_info *sockinfo;

    ev_io *io;  /* the stub variable to enable -fstrict-aliasing            */

    /*
     * for the SCTP alternative local addr may be invalid but not used    if so
     * put it here to emliate warnings.
     */
    tsock_ep *alter_ep = l_ep + 1;
    int i;                          /* the socket slot we check out         */

    char buf[INET6_ADDRSTRLEN_BUF];

    CHECKNULL(o_llc, return TRANS_A_NULL_POINTER);
    CHECKNULL(r_ep, return TRANS_A_NULL_POINTER);
    CHECKNULL(cb, return TRANS_A_NULL_POINTER);
    CHECKNULL(opts, return TRANS_A_NULL_POINTER);

    if(TRANS_INIT_MAGIC_NUM != trans_init_flag){
        return TRANS_NOT_INITIALIZE;
    }

    *o_llc = 0;

    if((opts->opts & SOCK_OPT_LOCAL_PORT) && (0 == is_ep_init(l_ep))){
        LOG(_CRI, "local port is passed in but local"
            "endpoint is not enabled, type(%s), port[%d], opts[%X].",
            prtl_to_str(l_ep->prtl), ntohs(l_ep->port), opts->opts);
        return TRANS_CROSSCHK_FAIL;
    }


    if(!(opts->opts & SOCK_OPT_CLOSE_ERR) && (NULL == cb->on_sock_err)){
        LOG(_CRI, "fail to go any furthur because "
            "SOCK_OPT_CLOSE_ERR is not set but cb is NULL.");
        return TRANS_CROSSCHK_FAIL;
    }

    if((SYS_SCTP == r_ep->prtl) && (sopts == NULL)){
        /* if it is a SCTP socket, with NULL opts will fail    */
        LOG(_CRI, "a SCTP endpoint with no SCTP options, will fail.");
        return TRANS_CROSSCHK_FAIL;
    }

    sockfd = t_socket(r_ep);
    if(sockfd < 0){
        LOG(_CRI, "create socket Fail in trans_connect, return[%d].", sockfd);
        return TRANS_FAILED;
    }

    CHKLOGLVL(L_RTM){
        buf[INET6_ADDRSTRLEN] = '\0';
        LOG(_RTM, "socket[%d] ==> Step 1, Creating socket for "
            "remote address[%s], port[%d], IP type(%s), protocol(%s).",
            sockfd,
            ep_to_addr(r_ep, buf, INET6_ADDRSTRLEN_BUF),
            ntohs(r_ep->port),
            type_to_str(r_ep->type), prtl_to_str(r_ep->prtl));
    }

    if(opts->opts & SOCK_OPT_LOCAL_PORT){

        tmp = 1;
        if(t_setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, &tmp, sizeof(int))){
            LOG(_ERR, "fail to set reuse addr to sock[%d], reason[%d](%s).",
                sockfd, errno, strerror(errno));
            t_close(sockfd);
            return TRANS_FAILED;
        }

        /* the local port is set, and it is a multicast remote endpoint
         * it means the local addr is used to send and recv messages
         *
         * try to bind on the local address, for Solaris, it may not work.
         * then try to bind on LOOPBACK.
         */
        if(is_multicast(r_ep)){
            if(TRANS_SUECCESS != t_bind(sockfd, r_ep)){
                LOG(_CRI, "socket[%d] bind local IP(m-cast) failed.", sockfd);
                t_close(sockfd);
                return TRANS_FAILED;
            }

            if(TRANS_FAILED == _set_multicast(sockfd, l_ep, r_ep,
                opts->opts & SOCK_OPT_MCAST_LOP)){
                LOG(_CRI, "fail to set multicast for loopback, quit.");
                t_close(sockfd);
                return TRANS_FAILED;
            }

        }else{
            /* bind the local IP address */
            if(TRANS_SUECCESS != t_bind(sockfd, l_ep)){
                LOG(_CRI, "socket[%d] bind local IP failed.", sockfd);
                t_close(sockfd);
                return TRANS_FAILED;
            }
            LOG(_RTM, "socket[%d] ==> Step 1, bind local address", sockfd);

            if((SYS_SCTP == l_ep->prtl) &&
                (opts->opts & SOCK_OPT_SCTP_ALTER)){

                if(0 == is_ep_init(alter_ep)){
                    LOG(_CRI, "the 2ed endpoint is invalid.");
                    t_close(sockfd);
                    return TRANS_FAILED;
                }

                /*
                 * try to bind an alternative endpoint for sctp, if it is
                 * correct save it in the sctp_opts in sockopts (if have)
                 */
                if(TRANS_SUECCESS != t_bind(sockfd, alter_ep)){
                    LOG(_CRI, "socket[%d] bind 2ed local IP failed", sockfd);
                    t_close(sockfd);
                    return TRANS_FAILED;
                }

            }
        }
    }

    if(opts->opts & SOCK_OPT_NON_BLOCKING){
        tmp = 1;
        if(ioctl(sockfd, FIONBIO, &tmp)){
            LOG(_CRI, "fail to ioctl non-blocking socket[%d], errno[%d](%s).",
                sockfd, errno, strerror(errno));
            t_close(sockfd);
            return TRANS_FAILED;
        }
    }

    tmp = SOCKET_BUF_SIZE;
    if(opts->opts & SOCK_OPT_SEDBUF){

        if(opts->sbuf_size > SOCKET_BUF_SIZE){
            /* have a mini send buf that is 128k */
            tmp = opts->sbuf_size;
        }

        if(t_setsockopt(sockfd, SOL_SOCKET, SO_SNDBUF, &tmp, sizeof(int))){

            LOG(_ERR, "fail to set the send buffer for "
                "socket [%d], errno[%d](%s)",
                sockfd, errno, strerror(errno));
            /* continue */
        }
    }

    if(opts->opts & SOCK_OPT_RCVBUF){
        if(opts->rbuf_size > SOCKET_BUF_SIZE){
            /* have a mini send buf that is 128k */
            tmp = opts->rbuf_size;
        }
        if(t_setsockopt(sockfd, SOL_SOCKET, SO_RCVBUF, &tmp, sizeof(int))){
            LOG(_ERR, "fail to set the recv buffer for "
                "socket [%d], errno[%d](%s)",
                sockfd, errno, strerror(errno));
            /* continue */
        }
    }

    /* turn off the linger, and should be turn on/off in connect or accept
     * based on opts */

    if(SYS_SCTP == r_ep->prtl){
        (void)_set_linger(sockfd, 1, opts->linger_time);
    }else{
        /* actually, it is not preferred to wait forever for TCP
         * however, turn of immediately will solve lots of problem */
        (void)_set_linger(sockfd, 0, 0);
    }

    tmp = 1;
    if(t_setsockopt(sockfd, SOL_SOCKET, SO_KEEPALIVE, &tmp, sizeof(int))){

        LOG(_ERR, "fail to turn off the keep-alive for "
            "socket [%d], errno[%d](%s).",
            sockfd, errno, strerror(errno));
        /* continue */
    }

    /* turn off the keep alive for TCP (for SCTP), 2 hours is no use */
    /*
     * XXX to enable it and give an interface to set the heartbeat
     */
    tmp = 0;
    if(t_setsockopt(sockfd, SOL_SOCKET, SO_KEEPALIVE, &tmp, sizeof(int))){

        LOG(_ERR, "fail to turn off the stupid keep-alive"
            " socket [%d], errno[%d](%s)",
            sockfd, errno, strerror(errno));
    }

    if(opts->opts & SOCK_OPT_NON_BLOCKING){
        tmp = 1;
        if(ioctl(sockfd, FIONBIO, &tmp)){
            LOG(_CRI, "fail to ioctl the non-blocking"
                " socket [%d], errno[%d](%s)",
                sockfd, errno, strerror(errno));
            t_close(sockfd);
            return TRANS_FAILED;
        }
    }

    if((SYS_SCTP == r_ep->prtl) && (cb->on_sctp_event)){
        if(TRANS_SUECCESS != _set_sctpsock_opt(sockfd)){
            t_close(sockfd);
            return TRANS_FAILED;
        }

        if(TRANS_SUECCESS != _set_sctpopt(sockfd, sopts)){
            LOG(_CRI, "failed to set the sctp opts");
            t_close(sockfd);
            return TRANS_FAILED;
        }
    }

    if(SYS_TCP == r_ep->prtl || SYS_SCTP == r_ep->prtl){

        if(opts->opts & SOCK_OPT_NDLAY){
            tmp = _set_no_delay(sockfd, r_ep->prtl, 1);
            if(TRANS_SUECCESS != tmp){
                LOG(_CRI, "fail to set the no-delay mode for "
                    "socket fd[%d], protocol(%s), opts[%X], res[%d](%s).",
                    sockfd, prtl_to_str(r_ep->prtl),
                    opts->opts, tmp, ERROR_TO_STR(tmp));
                t_close(sockfd);
                return TRANS_FAILED;
            }
        }
    }

    if(opts->opts & SOCK_OPT_SCTP_ALTER){
        tmp = 2;
    }else{
        tmp = 1;
    }

    /* This might be a blocking connect but soon it will timeout */
    do{
        conn_ret = t_connect(sockfd, r_ep, tmp, &err);
    }while(0 != conn_ret && err == EINTR);

    if(0 != conn_ret){

        /*
         * Connect failed due to other reason for example
         * EINPROGRESS or EWOULDBLOCK will continue to start re
         * connect timer depends on SOCK_OPT_CLOSE_ERR option
         */

        switch(err){
            case EINPROGRESS:
            case EWOULDBLOCK:
                LOG(_CHK, "connecting errno[%d]"
                    " and non-blocking flag[%d].", err,
                    opts->opts & SOCK_OPT_NON_BLOCKING);

                break;
            default:
                LOG(_CRI, "connect error %d", err);
                t_close(sockfd);
                return TRANS_FAILED;
        }

        if(0 == (opts->opts & SOCK_OPT_NON_BLOCKING)){
            LOG(_ASS, "a non-blocking in error socket[%d], err[%d]."
                , sockfd, err);
        }

        /* let it continue to start a timeout timer to do clearance */
        if(timeout > 0){
            need_clean = 1;
        }else{
            LOG(_CRI, "fail to connect socket[%d] and don't set "
                "a reconnect timer, err[%d], timeout[%d]."
                , sockfd, err, timeout);
            t_close(sockfd);
            return TRANS_FAILED;
        }

        LOG(_RTM, "socket[%d] ==> Step 2, Connecting "
                "failed, restart the timer.", sockfd);

    }else{
        LOG(_RTM, "socket[%d] ==> Step 2, Connecting "
                "to remote sueccessfully.", sockfd);
    }

    if(NULL == llc){
        info = _chkout_thread_pool();
        /* apparently, info can not be NULL */

    }else{
        info = ((socket_info *)llc)->thread;
    }

    pthread_mutex_lock(&info->mutex);

    i = _chkout_socket(info);
    if(i < 0){
        t_close(sockfd);
        pthread_mutex_unlock(&info->mutex);
        LOG(_CRI, "can not find an available socket"
            "slot for thread[%d], reason(%s).",
            (int)(info - m_info),
            trans_strerror(i));
        return TRANS_NOT_FOUND;
    }
    info->active_sock++;

    /* get a free socket slot, save it and set it */
    sockinfo = &info->psock_info[i];

    /* XXX Does -O2 will enroll this memcpy? I guess not */
    /* memcpy(&sockinfo->opts, opts, sizeof(sock_opts)); */
    sockinfo->opts.ip_tos = opts->ip_tos;
    sockinfo->opts.linger_time = opts->linger_time;
    sockinfo->opts.opts = opts->opts;
    sockinfo->opts.rbuf_size = opts->rbuf_size;
    sockinfo->opts.sbuf_size = opts->sbuf_size;
    sockinfo->thread = info;
    sockinfo->ulc = ulc;
    sockinfo->prtl = r_ep->prtl;
    memcpy(&sockinfo->cb, cb, sizeof(sock_cb));
    sockinfo->fd = sockfd;

    /* if it is a SCTP, we need to init the sock_info sctp opt */
    if(SYS_SCTP == r_ep->prtl){
        sockinfo->sctp_opts = malloc(sizeof(sctp_opts));
        if(NULL == sockinfo->sctp_opts){
            /* no need return the socket slot because it is still IDEL */
            t_close(sockfd);
            info->active_sock--;
            pthread_mutex_unlock(&info->mutex);
            LOG(_CRI, "failed to alloc memory for sctp opts.");
            return TRANS_NOT_ENOUGH_MEM;
        }

        /* sopts is definately not NULL */
        memcpy(sockinfo->sctp_opts, sopts, sizeof(sctp_opts));
    }

    if(need_clean){

        sockinfo->state = S_CONNECTING;
        /* if timeout is set >0, an internal timer is started to clean */
        memcpy(&sockinfo->r_ep, r_ep, sizeof(tsock_ep));

        if(TRANS_SUECCESS != __set_sock_timer(sockinfo, TIMER_TYPE_INTERNAL,
            __socket_timer_cb, timeout)){

            _reset_socket_info(sockinfo, 0);
            info->active_sock--;
            t_close(sockfd);
            pthread_mutex_unlock(&info->mutex);
            LOG(_CRI, "failed to set the reconect timer.");
            return TRANS_FAILED;
        }
    }else{
        if(opts->opts & SOCK_OPT_LOCAL_PORT){
            if(is_multicast(r_ep)){
                /* save the remote address */
                memcpy(&sockinfo->r_ep, r_ep, sizeof(tsock_ep));
            }
            memcpy(&sockinfo->l_ep, l_ep, sizeof(tsock_ep));
        }else{
            if(TRANS_SUECCESS != _save_conn_addr(sockinfo)){
                _reset_socket_info(sockinfo, 0);
                info->active_sock--;
                t_close(sockfd);
                pthread_mutex_unlock(&info->mutex);
                LOG(_CRI, "failed to set the address.");
                return TRANS_FAILED;
            }
        }
        sockinfo->state =S_CONNECT;
    }

    *o_llc = (void*) sockinfo;

    /*
     * the reason why we use this temp variable stub is because the
     * compiler will do the strict aliasing optimization for target
     * independant porting. if we don't do this, we can
     *         1, disable optimization by using -fno-strict-aliasing
     *         2, use __attribute__((__may_alias__)) wrapper
     *         3, ignore by silent
     *     so it will cause program target-dependant.
     */
    io = &sockinfo->ev;
    /* add for the libev */
    ev_init(io, __sock_cb_entry);
    sockinfo->ev.data = (void*)sockinfo;

    if(need_clean){
        /* The retry timer will check the EV_WRITE status */
        ev_io_set(io, sockinfo->fd, EV_READ | EV_WRITE);
    }else{
        /* because it is now SOCK_CONNECT, try to read on receive */
        ev_io_set(io, sockinfo->fd, EV_READ);
    }
    ev_io_start(info->base_loop, io);

    pthread_mutex_unlock(&info->mutex);

    LOG(_RTM, "socket[%d] ==> Step 3 (DONE), Event added "
            "sueccessfully, for thread[%d], slot[%d]."
            , sockfd, (int)(info - m_info), i);

    __wakeup_thread(info);

    return TRANS_SUECCESS;
}

int trans_send(void *llc, char* msg, int len, tsock_ep* ep){

    register thread_info *tinfo;
    register socket_info *sinfo = (socket_info *)llc;
    int res;

    CHECKNULL(llc, return TRANS_A_NULL_POINTER);
    CHECKNULL(msg, return TRANS_A_NULL_POINTER);

    if(len <= 0){
        LOG(_ASS, "passing an invalid len[%d].", len);
        return TRANS_INVALID_VALUE;
    }

    if(ep){
        if(SYS_UDP != ep->prtl){
            LOG(_ASS, "try to call UDP send but the remote peer protocol"
                "is not UDP, now(%s).", prtl_to_str(ep->prtl));
            return TRANS_CROSSCHK_FAIL;
        }
    }

    tinfo = sinfo->thread;

    /* Check before dereference */
    if(S_CONNECT != sinfo->state && S_TRANSIENT != sinfo->state){
        LOG(_ERR, "the connection state(%s) is not correct.",
            _stat_to_str(sinfo->state));
        return 0;
    }

    CHECKNULL(tinfo, return TRANS_A_NULL_POINTER);
    pthread_mutex_lock(&tinfo->mutex);

    if(S_CONNECT != sinfo->state && S_TRANSIENT != sinfo->state){
        pthread_mutex_unlock(&tinfo->mutex);
        LOG(_ERR, "the connection state(%s) is not correct.",
            _stat_to_str(sinfo->state));
        return 0;
    }

    res = __trans_send(llc, msg, len, ep);
    if(res < 0){
        if((sinfo->cb.on_sock_err) &&
            0 == (sinfo->opts.opts & SOCK_OPT_CLOSE_ERR)){

            pthread_mutex_unlock(&tinfo->mutex);
            (void)sinfo->cb.on_sock_err(sinfo, sinfo->ulc, TRANS_FAILED);
            pthread_mutex_lock(&tinfo->mutex);
        }

        sinfo->state = S_CLOSING;

        if(sinfo->cb.on_close){

            pthread_mutex_unlock(&tinfo->mutex);
            (void)sinfo->cb.on_close(sinfo, sinfo->ulc, TRANS_FAILED);
            pthread_mutex_lock(&tinfo->mutex);
        }

    }else{
        sinfo->num_send++;
    }
    pthread_mutex_unlock(&tinfo->mutex);
    return res;
}

int trans_close(void *llc){

    register int res;
    register thread_info *thread;

    CHECKNULL(llc, return TRANS_A_NULL_POINTER);

    thread = ((socket_info *)llc)->thread;

    pthread_mutex_lock(&thread->mutex);
    res = __trans_close((socket_info *)llc);
    pthread_mutex_unlock(&thread->mutex);
    __wakeup_thread(thread);

    return res;
}

int trans_set_timer(void *ulc, void **tlc, int id, int m_sec, timer_cb cb){

    int res;
    socket_info info;

    info.thread = &m_info[0];

    CHECKNULL(tlc, return TRANS_A_NULL_POINTER);
    CHECKNULL(cb, return TRANS_A_NULL_POINTER);

    if(TRANS_INIT_MAGIC_NUM != trans_init_flag){
        LOG(_ASS, "trans not initilized yet.");
        return TRANS_NOT_INITIALIZE;
    }

    pthread_mutex_lock(&info.thread->mutex);
    res = __set_timer(&info, tlc, ulc, id, m_sec, cb, TIMER_TYPE_SIMPLE);
    pthread_mutex_unlock(&info.thread->mutex);

    /* this is a must because the thread might be in sleep */
    __wakeup_thread(info.thread);
    return res;
}

int trans_set_timer_ex(void *tlc, void* ulc, void **t_tlc, int id, int m_sec,
    timer_cb cb, int type){

    int res;
    socket_info info;

    CHECKNULL(t_tlc, return TRANS_A_NULL_POINTER);
    CHECKNULL(cb, return TRANS_A_NULL_POINTER);

    if(NULL == tlc){
        info.thread = &m_info[0];
    }else{
        info.thread = ((timer_info *)tlc)->thread;
    }

    info.t_llc = t_tlc;

    if(NULL == cb){
        LOG(_ASS, "bad parameter passed in, ulc[%p], cb[%p]", ulc, cb);
        return TRANS_FAILED;
    }

    if(TRANS_INIT_MAGIC_NUM != trans_init_flag){
        LOG(_ASS, "trans not initilized yet.");
        return TRANS_NOT_INITIALIZE;
    }

    pthread_mutex_lock(&info.thread->mutex);
    res = __set_timer(&info, t_tlc, ulc, id, m_sec, cb, type);
    pthread_mutex_unlock(&info.thread->mutex);

    /* this is a must because the thread might be in sleep */
    __wakeup_thread(info.thread);
    return res;

}

int trans_kill_timer(void *tlc, int id){

    int res;
    register thread_info *info;

    CHECKNULL(tlc, return TRANS_A_NULL_POINTER);

    info = ((timer_info *)tlc)->thread;

    CHECKNULL(info, return TRANS_A_NULL_POINTER);

    if(TRANS_INIT_MAGIC_NUM != trans_init_flag){
        LOG(_ASS, "trans not initilized yet.");
        return TRANS_NOT_INITIALIZE;
    }

    pthread_mutex_lock(&info->mutex);
    res = __kill_timer(tlc, id);
    pthread_mutex_unlock(&info->mutex);

    return res;
}

int trans_get_timer_remains(void *tlc){

    timer_info *timer = (timer_info *)tlc;
    ev_tstamp remaining;

    CHECKNULL(tlc, return 0);

    if(TRANS_INIT_MAGIC_NUM != trans_init_flag){
        LOG(_ASS, "trans not initilized yet.");
        return TRANS_NOT_INITIALIZE;
    }

    pthread_mutex_lock(&timer->thread->mutex);
    remaining = ev_timer_remaining(timer->thread->base_loop,
        &timer->ev);
    pthread_mutex_unlock(&timer->thread->mutex);

    return (int)(remaining * 1000);
}

const tsock_ep* trans_get_laddr(void *llc){
    CHECKNULL(llc, return NULL)
    return &((socket_info *)llc)->l_ep;
}

const tsock_ep* trans_get_raddr(void *llc){
    CHECKNULL(llc, return NULL)
    return &((socket_info *)llc)->r_ep;
}

int trans_get_conn(void *llc, tsock_ep* l, tsock_ep* r){

    register socket_info *info = (socket_info *)llc;
    CHECKNULL(llc, return TRANS_A_NULL_POINTER);
    CHECKNULL(l, return TRANS_A_NULL_POINTER);
    CHECKNULL(r, return TRANS_A_NULL_POINTER);

    memcpy(l, &info->l_ep, sizeof(tsock_ep));
    memcpy(r, &info->r_ep, sizeof(tsock_ep));

    return TRANS_SUECCESS;
}

void* trans_get_my_ulc(void *llc){
    CHECKNULL(llc, return NULL)
    return ((socket_info *)llc)->ulc;
}

pthread_t trans_get_tid(void *llc){
    CHECKNULL(llc, return pthread_self())
    return ((socket_info *)llc)->thread->my_tid;
}

/*
 * This is the trans thread loop entry functions,
 * The args, stores the thread_info structure.
 */
static void * __thread_loop(void* args){

    register thread_info *info = (thread_info *)args;

    ev_io *pio;

    LOG(_DBG, "trans Thread[%d] started.", info->index);

    /*
     * Create the pipe and watch it because
     *     1, if nothing watched, libev loop will return
     *     2, use the UNIX way to let libev wake up the thread gracefully.
     */
    pio = &info->p_ev;
    ev_init(pio, __pipe_callback);
    /* a very implict data */
    pio->data = args;
    /* set the io and for its watch type */

    ev_io_set(pio, info->fd[0], EV_READ | EV_WRITE);
    ev_io_start(info->base_loop, pio);

    /* libev main loop */
    LOG(_DBG, "ev is running in.");
    ev_run(info->base_loop, 0);

    /* free the resource at once */
    pthread_detach(info->my_tid);

    LOG(_RTM, "trans Thread[%d] is dead.", info->index);

    return NULL;
}

/*
 * This is the UNIX way of awake a thread, if we use the linux way
 * A pthread_cond_t should be included in the thread_info, but the
 * pipe is more graceful and effictive.
 *
 */
static int __create_pipe(thread_info *pinfo){

    if(-1 == pipe(pinfo->fd)){
        LOG(_CRI, "fail to create pipe for thread[%d].", pinfo->index);
        return TRANS_FAILED;
    }

    if(fcntl(pinfo->fd[0], F_SETFL, O_NONBLOCK)
                    || fcntl(pinfo->fd[1], F_SETFL, O_NONBLOCK)){
        LOG(_CRI, "fail to set pipes to non-blocking mode in thread [%d].",
            pinfo->index);

        return TRANS_FAILED;
    }

    return TRANS_SUECCESS;
}

static void __close_pipe(thread_info *pinfo){
    close(pinfo->fd[1]);
    close(pinfo->fd[0]);
}

/*
 * The pipe callback from the libev, we need to read all bytes to
 * clean the pipe. and the thread is awake because of it.
 */
static void __pipe_callback (struct ev_loop *loop, ev_io *w, int revents){

    register thread_info *info;
    register int res;

    char buf[16];

    info = (thread_info *)w->data;

    if(info && (revents | EV_READ)){
        /* read bytes to empty the thread pipe */
        while((res = read(info->fd[0], buf, sizeof(buf))) == sizeof(buf)){
            ;
        }
    }else{
        LOG(_WRN, "get an error event.");
    }

    return;
}

/*
 * This function wake up a thread, write a byte to fd[1] and fd[0]
 * receives.
 */
static int __wakeup_thread(thread_info *pinfo){

    pthread_t tid;
    char dummy[1] = {'\0'};

    register int res;

    tid = pthread_self();

    if(pinfo->my_tid == tid){
        /* already awake */
        return TRANS_SUECCESS;
    }

    res = write(pinfo->fd[1], dummy, sizeof(dummy));
    if(res != sizeof(dummy)){
        LOG(_ERR, "error when write pipe.");
        return TRANS_FAILED;
    }

    return TRANS_SUECCESS;
}


static void __timer_cb_entry(struct ev_loop *loop, ev_timer *w, int revents){

    /* of cource we can get the timer_info in the w and find the thread mutex */
    _ev_lock_acquire(loop);
    __timer_callback(loop, w, revents);
    _ev_lock_release(loop);

    /* put the thread to the scheduler */
    __nanosleep(10);

}
/*
 * This is the timer event dispatcher
 */
static void __timer_callback(struct ev_loop *loop, ev_timer *w, int revents){

    timer_info *tinfo = (timer_info *)(w->data);

    pthread_mutex_unlock(&tinfo->thread->mutex);
    tinfo->on_timer(tinfo->id, tinfo->llc, tinfo->ulc);
    pthread_mutex_lock(&tinfo->thread->mutex);

    if(tinfo->type & TIMER_TYPE_REPEAT){
        ev_timer_again(loop, w);
    }else{
        ev_timer_stop(loop, w);
        tinfo->state = T_IDLE;
        LIST_INSERT_HEAD(&tinfo->thread->tlist, tinfo, next);
    }
}

static int __set_sock_timer(socket_info *info, int id, timer_cb cb, int tout){

    if(info->t_llc){
        LOG(_ASS, "there's an unexpect append timer on socket.");
        CHKLOGLVL(L_ASS){
            _print_slot(info);
        }
        return TRANS_FAILED;
    }

    return __set_timer(info, &info->t_llc, info, id, tout, cb,
        TIMER_TYPE_SIMPLE);

}

static int __kill_sock_timer(socket_info *info){
    int res;

    if(NULL == info->t_llc){
        return TRANS_SUECCESS;
    }

    res = __kill_timer((timer_info *)info->t_llc, TIMER_TYPE_INTERNAL);
    info->t_llc = NULL;
    return res;
}

/*
 * The reason why we don't need an entry function to it is because
 * it is internal use, and will not be locked by trans.
 */
static int __socket_timer_cb(int id, void *llc, void *ulc){

    socket_info *p = (socket_info *)ulc;

    if(NULL == p){
        return TRANS_SUECCESS;
    }

    if(S_IDEL != p->state){
        /*
         * The socket is to be closed.
         * XXX this is a trick for the user callback close and open a new
         * socket, and when user checking out the same socket slot,
         * and there's some error it will brings.
         */
        p->state = S_CLOSING;
    }

    if(p->cb.on_conn_err){
        p->cb.on_conn_err(p, p->ulc, TRANS_TIME_OUT);
    }

    if(S_CLOSING == p->state){
        LOG(_RTM, "timeout for connect timer, socket will be closed");
        __trans_close(p);
    }

    /* the timer has been moved into the free list, then, detach */
    p->t_llc = NULL;
    return TRANS_SUECCESS;
}

/*
 * The main socket entry fuction and its dispatcher
 */

static void __sock_cb_entry(struct ev_loop *loop, ev_io *w, int revents){
    /* of cource we can get the timer_info in the w and find the thread mutex */
    _ev_lock_acquire(loop);
    __sock_callback(loop, w, revents);
    _ev_lock_release(loop);

}
static void __sock_callback(struct ev_loop *loop, ev_io *w, int revents){

    register socket_info *sockinfo = (socket_info *)w->data;
    int error_occured = 0;
    int error = TRANS_FAILED;
    socklen_t error_len = sizeof(error);
    int tmp;

    int accept_fd;
    tsock_ep accept_ep;
    register thread_info *accept_tinfo = NULL;
    int i; /* the socket slot we check out */
    register socket_info *accept_sinfo;

    int recv_len;

    ev_io *io;

    char buf[INET6_ADDRSTRLEN_BUF];

    switch(sockinfo->state){
        case S_CONNECTING:
            if(revents | EV_ERROR){
                /*
                 * This is not an ev_error but a SO_ERROR, if checked by
                 * t_getsockopt SOL_SOCKET. finally, the error will be zero
                 *
                 * which means, the socket is connected.
                 *
                 * can NOT break because it is write still need to do.
                 */
                LOG(_DBG, "the client is retrying.");
                /* break */
            }

            /*
             * For TCP, when doing EV_WRITE, should be doing a handshake
             * and will turn to the SOCK_CONNECT if sueccessfully
             *
             * For SCTP, it is much more complex, we need to handle the
             * EV_READ for SCTP status change.
             */
            if((revents | EV_WRITE)        /* TCP and then SCTP */
                || ((revents | EV_READ) && (SYS_SCTP == sockinfo->prtl))){

                if(t_getsockopt(sockinfo->fd, SOL_SOCKET, SO_ERROR,
                    &error, &error_len) || error){
                    LOG(_CRI, "error happened when turning IDEL to CONNECT "
                        "for socket fd[%d], res[%d], errno[%d](%s).",
                        sockinfo->fd, error, errno, strerror(errno));
                    /* error is set */
                    error_occured = 1;
                    break;
                }

                if(SYS_SCTP == sockinfo->prtl){
                    /*
                     * TODO SCTP
                     * check every SCTP step to continue step by step
                     *
                     * read from the socket for SCTP notification to change
                     */
                    error_occured = 1;
                    error = TRANS_UNSUPPORTED;
                    break;
                }

                /* Update the event in libev */
                ev_io_stop(loop, w);

                /* don't need to reset the ev by ev_init because the
                 * sockinfo->ev_r.data is still sockinfo itself.
                 */

                /* Changing state for socket */
                sockinfo->state = S_CONNECT;
                /* Only care about the recv  */
                ev_io_set(&sockinfo->ev, sockinfo->fd, EV_READ);
                ev_io_start(loop, &sockinfo->ev);

                CHKLOGLVL(L_RTM){
                    buf[INET6_ADDRSTRLEN] = '\0';
                    LOG(_RTM, "the socket[%d] is eventually connected "
                        "to address(%s), port[%d].",
                        sockinfo->fd,
                        ep_to_addr(&sockinfo->r_ep, buf, INET6_ADDRSTRLEN_BUF),
                        ntohs(sockinfo->r_ep.port));
                }

                /*
                 * Invoke the callback
                 *
                 * Use the socket thread to guard the state change before,
                 * now need the thread mutex to guard the callback.
                 */
                if(sockinfo->cb.on_connect){
                    if(sockinfo->t_llc){
                        (void)__kill_sock_timer(sockinfo);
                    }
                    /* save the address, no return check */
                    (void)_save_conn_addr(sockinfo);
                    pthread_mutex_unlock(&sockinfo->thread->mutex);
                    sockinfo->cb.on_connect((void*)sockinfo, sockinfo->ulc);
                    pthread_mutex_lock(&sockinfo->thread->mutex);
                }

            }

            break;
        case S_LISTEN:

            if(revents & EV_READ){

                if(SYS_SCTP == sockinfo->prtl){
                    /*
                     * TODO SCTP
                     *
                     * Read from the socket,
                     * to determine if it is a notification
                     */
                    error_occured = 1;
                    error = TRANS_UNSUPPORTED;
                    break;
                }else{
                    /*
                     * only TCP can hit here,
                     *
                     * Although it will blocked here, it won't last long
                     * because there's already a write opration, the accept
                     * will return quickly.
                     */
                    accept_fd = t_tcp_accept(sockinfo->fd, &accept_ep);
                    if(accept_fd < 0){
                        error_occured = 1;
                        error = errno;
                        break;
                    }

                    if(sockinfo->opts.opts & SOCK_OPT_LISTEN_ONCE){
                        ev_io_stop(loop, w);
                        t_close(sockinfo->fd);
                        sockinfo->thread->active_sock--;
                        /*
                         * issue 1, LISTEN_ONCE and when accept a new link
                         * if reset the sockinfo, will cause a core dump below
                         *
                         * No need to reset the sockinfo but need to
                         * 1, introduce a new state S_LISTEN_DONE
                         * 2, __trans_close to clean the socket info
                         * 3, for double check, just reset the fd to -1;
                         *
                         *
                        _reset_socket_info(sockinfo, 0);
                        */
                        LOG(_CHK, "socket[%d] listen on port[%d] will be "
                            "closed because listen is done ONCE.",
                            sockinfo->fd, ntohs(sockinfo->l_ep.port));

                        sockinfo->state = S_LISTEN_DONE;
                        sockinfo->fd = -1;

                    }else{

                        accept_tinfo = _chkout_thread_pool();
                        if(NULL == accept_tinfo){
                            error_occured = 1;
                            error = TRANS_NOT_FOUND;
                            LOG(_CRI, "no available thread pool.");
                            break;
                        }

                        if(accept_tinfo == sockinfo->thread){
                            /*
                             * XXX A trick to simpify the following work. if it
                             * is not NULL, will try to acquire the thread
                             * again and cause a dead lock
                             *
                             * So here set it to NULL to let program to see it.
                             */
                            accept_tinfo = NULL;
                        }
                    }

                    if(accept_tinfo){
                        pthread_mutex_lock(&accept_tinfo->mutex);

                        i = _chkout_socket(accept_tinfo);
                        if(i < 0){
                            pthread_mutex_unlock(&accept_tinfo->mutex);
                            LOG(_CRI, "can not find an available socket"
                                "slot for thread[%d], reason(%s).",
                                (int)(accept_tinfo - m_info),
                                trans_strerror(i));
                            error_occured = 1;
                            error = TRANS_NOT_FOUND;
                            break;
                        }

                        accept_tinfo->active_sock++;
                        accept_sinfo = &accept_tinfo->psock_info[i];
                        /* The lock will be released later */
                    }else{
                        i = _chkout_socket(sockinfo->thread);
                        if(i < 0){
                            LOG(_CRI, "can not find an available socket"
                                "slot for thread[%d], reason(%s).",
                                (int)(sockinfo->thread - m_info),
                                trans_strerror(i));
                            error_occured = 1;
                            error = TRANS_NOT_FOUND;
                            break;
                        }

                        sockinfo->thread->active_sock++;
                        accept_sinfo = &sockinfo->thread->psock_info[i];
                    }

                    accept_sinfo->opts.ip_tos = sockinfo->opts.ip_tos;
                    accept_sinfo->opts.linger_time =
                                    sockinfo->opts.linger_time;
                    accept_sinfo->opts.opts = sockinfo->opts.opts;
                    accept_sinfo->opts.rbuf_size = sockinfo->opts.rbuf_size;
                    accept_sinfo->opts.sbuf_size = sockinfo->opts.sbuf_size;

                    if(accept_tinfo){
                        accept_sinfo->thread = accept_tinfo;
                    }else{
                        accept_sinfo->thread = sockinfo->thread;
                    }

                    accept_sinfo->ulc = sockinfo->ulc;
                    accept_sinfo->prtl = sockinfo->prtl;
                    memcpy(&accept_sinfo->cb, &sockinfo->cb, sizeof(sock_cb));
                    accept_sinfo->fd = accept_fd;

                    /*
                     * TODO SCTP
                     * malloc and save the SCTP opt
                     * if it is a SCTP, we need to init the sock_info sctp opt
                     */

                    (void)_save_conn_addr(accept_sinfo);
                    if(SYS_TCP == accept_sinfo->prtl){
                        if(_set_no_delay(accept_fd, SYS_TCP, 1)){
                            LOG(_ERR, "fail to set nodelay.");
                            /* continue */
                        }
                    }

                    if(accept_sinfo->opts.opts & SOCK_OPT_NON_BLOCKING){
                        tmp = 1;
                        if(ioctl(accept_fd, FIONBIO, &tmp)){
                            LOG(_CRI, "fail to ioctl the "
                                "non-blocking socket[%d], errno[%d](%s).",
                                accept_fd, errno, strerror(errno));

                            error = errno;
                            t_close(accept_fd);

                            if(accept_tinfo){
                                accept_tinfo->active_sock--;
                                _reset_socket_info(accept_sinfo, 0);
                                pthread_mutex_unlock(&accept_tinfo->mutex);
                            }else{
                                _reset_socket_info(accept_sinfo, 0);
                                sockinfo->thread->active_sock--;
                            }

                            error_occured = 1;
                            break;
                        }
                    }

                    io = &accept_sinfo->ev;

                    ev_init(io, __sock_cb_entry);
                    accept_sinfo->ev.data = (void*)accept_sinfo;
                    ev_io_set(io, accept_sinfo->fd, EV_READ);

                    if(accept_tinfo){
                        ev_io_start(accept_tinfo->base_loop, io);
                        pthread_mutex_unlock(&accept_tinfo->mutex);
                    }else{
                        ev_io_start(sockinfo->thread->base_loop, io);
                    }

                    pthread_mutex_unlock(&sockinfo->thread->mutex);

                    tmp = sockinfo->cb.on_new(accept_sinfo, sockinfo->ulc,
                        (void*)&accept_sinfo->ulc, &accept_ep);

                    if(accept_tinfo){
                        pthread_mutex_lock(&accept_tinfo->mutex);
                    }
                    pthread_mutex_lock(&sockinfo->thread->mutex);

                    /* TODO and XXX     carefully check here */
                    /* client may closed the link in callback */
                    if(S_IDEL == accept_sinfo->state){
                        if(accept_tinfo){
                            pthread_mutex_unlock(&accept_tinfo->mutex);
                        }
                        break;
                    }

                    /* Check the return value of the on_new call */
                    if(APP_ACCEPT == tmp){

                        accept_sinfo->state = S_CONNECT;
                        CHKLOGLVL(L_RTM){
                            _print_slot(accept_sinfo);
                        }

                    }else if(APP_REJECT == tmp){

                        LOG(_CHK, "the new connection is "
                            "rejected by server, closing.");

                        ev_io_stop(accept_sinfo->thread->base_loop, io);
                        accept_sinfo->thread->active_sock--;
                        t_close(accept_fd);
                        _reset_socket_info(accept_sinfo, 0);

                        if(accept_tinfo){
                            pthread_mutex_unlock(&accept_tinfo->mutex);
                        }
                        break;

                    }else if(APP_IGNORE == tmp){
                        /*
                         * Why would it open a socket for listen but ignore
                         * EVERY connection?
                         *
                         * So it is not implemented.
                         *
                         * If need, it resets the IO event to EV_ERROR
                         */
                        LOG(_ASS, "try to ignore listening"
                            ", but libtrans is not support it yet.");

                        /* ignore by silent */
                        if(accept_tinfo){
                            pthread_mutex_unlock(&accept_tinfo->mutex);
                        }
                        break;

                    }else{
                        LOG(_ASS, "the user returns an invalid value!");

                        /* ignore by silent */
                        if(accept_tinfo){
                            pthread_mutex_unlock(&accept_tinfo->mutex);
                        }
                        break;
                    }

                    /*
                     * Try to wake up the thread if the new thread
                     *
                     * ONLY the case APP_ACCEPT will hit here.
                     */
                    if(accept_tinfo){
                        pthread_mutex_unlock(&accept_tinfo->mutex);
                        __wakeup_thread(accept_tinfo);
                    }
                }
            }

            if(revents & EV_ERROR){
                LOG(_ASS, "POLLERR | POLLHUP happened.");
                error = TRANS_LIBEV_ERROR;
                error_occured = 1;
                break;
            }

            break;
        case S_CONNECT:

            if(revents & EV_WRITE){
                /*
                 * Don't support yet because the block queue is not supported.
                 *
                 * It assume that the socket is writable so the select() is not
                 * called.
                 *
                 * TODO block queue for each socket.
                 *
                 * If the message is stored in the send queue, should do
                 *     1, Flush the queue, try to send as much as possible
                 *     2, If queue is flushed, give user a notification.
                 *     3, any bad idea?
                 */
                LOG(_WRN, "don't support yet.");
                /* break;    */
            }

            if(revents & EV_READ){
                /* reuse the accept_ep     */
                tmp = __read_socket(sockinfo, &recv_len, &accept_ep);

                if(0 == tmp){
                    /* not a full package */
                    break;
                }else if(tmp < 0){
                    error = tmp;
                    error_occured = 1;
                    break;
                }else{

                    switch(sockinfo->prtl){
                        case SYS_UDP:
                            if(!is_ep_init(&accept_ep) ||
                                NULL == sockinfo->cb.on_read_udp){
                                break;
                            }

                            pthread_mutex_unlock(&sockinfo->thread->mutex);
                            sockinfo->cb.on_read_udp(sockinfo, sockinfo->ulc,
                                &accept_ep, &sockinfo->l_ep,
                                sockinfo->recv_buf, tmp);
                            pthread_mutex_lock(&sockinfo->thread->mutex);
                            break;
                        case SYS_TCP:
                            pthread_mutex_unlock(&sockinfo->thread->mutex);
                            sockinfo->cb.on_read_tcp(sockinfo, sockinfo->ulc,
                                sockinfo->recv_buf, tmp);
                            pthread_mutex_lock(&sockinfo->thread->mutex);
                            break;
                        case SYS_SCTP:
                            break;
                        default:
                            break;
                    }
                }
            }

            if(revents & EV_ERROR){
                LOG(_ASS, "POLLERR | POLLHUP on connect socket.");
                error = TRANS_LIBEV_ERROR;
                error_occured = 1;
                break;
            }

            break;

        case S_CLOSING:

            LOG(_CRI, "the socket[%d] is in pending close.", sockinfo->fd);
            CHKLOGLVL(L_CRI){
                _print_slot(sockinfo);
            }
            error_occured = 1;
            break;

        default:
            /* Close the event to protect libtrans from being attacked */
            LOG(_ASS, "unknow socket state(%s).",
                _stat_to_str(sockinfo->state));
            CHKLOGLVL(L_ASS){
                _print_slot(sockinfo);
            }
            error_occured = 1;
            break;

    }

    if(error_occured){
        switch(sockinfo->state){
            case S_CONNECTING:

                if(sockinfo->t_llc){
                    __kill_sock_timer(sockinfo);
                }

                sockinfo->state = S_CLOSING;
                if(sockinfo->cb.on_conn_err){
                    pthread_mutex_unlock(&sockinfo->thread->mutex);
                    sockinfo->cb.on_conn_err(sockinfo, sockinfo->ulc, error);
                    pthread_mutex_lock(&sockinfo->thread->mutex);
                }

                break;
            case S_LISTEN:

                if(sockinfo->cb.on_listen_err){
                    pthread_mutex_unlock(&sockinfo->thread->mutex);
                    sockinfo->cb.on_listen_err(sockinfo, sockinfo->ulc, error);
                    pthread_mutex_lock(&sockinfo->thread->mutex);
                }

                break;
            case S_CONNECT:

                if((sockinfo->cb.on_sock_err) &&
                    0 == (sockinfo->opts.opts & SOCK_OPT_CLOSE_ERR)){

                    pthread_mutex_unlock(&sockinfo->thread->mutex);
                    sockinfo->cb.on_sock_err(sockinfo, sockinfo->ulc, error);
                    pthread_mutex_lock(&sockinfo->thread->mutex);
                }

                sockinfo->state = S_CLOSING;

                if(sockinfo->cb.on_close){

                    pthread_mutex_unlock(&sockinfo->thread->mutex);
                    sockinfo->cb.on_close(sockinfo, sockinfo->ulc, error);
                    pthread_mutex_lock(&sockinfo->thread->mutex);
                }

                break;
            case S_CLOSING:
                /* not closed by __sock_callback, close it now */
                __trans_close(sockinfo);
                break;
            default:
                __trans_close(sockinfo);
                break;
        }
    }

}

static int __set_timer(socket_info *info, void** outllc, void *ulc,
    int id, int mill_sec, timer_cb cb, int type){

    register timer_info *tinfo;
    float realtime;

    ev_timer *timer; /* the stub variable to enable -fstrict-aliasing */

    /* this is the return llc                       */
    *outllc = (void*)0;

    /* if there's no timer entry, create a new one  */
    if(LIST_EMPTY(&info->thread->tlist)){
        /* alloc one entry for the timer */
        tinfo = (timer_info *)malloc(sizeof(timer_info));
        if(NULL == tinfo){
            LOG(_CRI, "failed to alloc a timer node.");
            return TRANS_NOT_ENOUGH_MEM;
        }
    }else{
        /* else check out the first and unlink it   */
        tinfo = LIST_FIRST(&info->thread->tlist);
        LIST_REMOVE(tinfo, next);
    }

    tinfo->ulc = ulc;
    tinfo->id = id;
    tinfo->state = T_INUSE;
    tinfo->type = type;
    tinfo->llc = tinfo;

    if(NULL == cb){
        tinfo->on_timer = info->cb.on_default_timer;
    }else{
        tinfo->on_timer = cb;
    }
    /* set timer info to the socket pointer */
    tinfo->thread = info->thread;

    /*
     * This useless because sinfo might be a tmp variable
     * it is the thread the timer needed
    sinfo->timer_llc = (void*)tinfo;
    */

    /* set the output */
    *outllc = (void*)tinfo;
    realtime = mill_sec / 1000;

    timer = &tinfo->ev;
    ev_timer_init(timer, __timer_cb_entry, realtime, 0.);
    tinfo->ev.data = (void*)tinfo;

    ev_timer_start(tinfo->thread->base_loop, &tinfo->ev);

    return TRANS_SUECCESS;
}

static int __kill_timer(timer_info *timer, int id){

    register thread_info *tinfo;


#ifdef _T_DEBUG
    timer_info *timer_d;
#endif

    CHECKNULL(timer, return TRANS_A_NULL_POINTER);

    tinfo = timer->thread;

    CHECKNULL(tinfo, return TRANS_A_NULL_POINTER);

    /* If the timer is already killed, user may killed it
     * more times and it is safe
     */
    if(T_IDLE == timer->state){
        LOG(_CHK, "timer already killed.");

#ifdef _T_DEBUG
        /*
         * XXX we need to set a debug here just in case leaks
         * If the timer is idle, it should be in the free list
         */
        LIST_FOREACH(timer_d, &tinfo->tlist, next){
            if(timer_d == timer){
                return TRANS_SUECCESS;
            }
        }
        return TRANS_NOT_FOUND;
#else
        return TRANS_SUECCESS;
#endif
    }

    if(id != timer->id){
        LOG(_CRI, "timer id[%d] not found.", id);
        return TRANS_NOT_FOUND;
    }

    ev_timer_stop(tinfo->base_loop, &timer->ev);
    timer->state = T_IDLE;

    LIST_INSERT_HEAD(&tinfo->tlist, timer, next);

    return TRANS_SUECCESS;
}

static void __nanosleep(ulong msec){

    struct timespec rqtp, rmtp;

    rqtp.tv_sec = msec / 1000;
    msec %= 1000;
    if(0 == msec){
        rqtp.tv_nsec = 0;
    }
    else{
        rqtp.tv_nsec = 1000000 * msec;
    }
    (void)nanosleep(&rqtp, &rmtp);
    return;
}

/*
 * The read socket function has a very complex implemetation.
 *
 * Note:
 *         1, Support TCP and UDP plain data.
 *         2, Don't support any extension protocol for TCP such as TPKT.
 *         3, TODO list
 *             a) TLS, DTLS, binary IO stream. need openssl.
 *             b) IKEv2 or even IPSEC, need openswan
 *             c) SCTP
 *             d) BULK read, read a large area of data...
 *
 * It stores the read info in the socket_info's recv_buf
 * when the package is ready, it will return the whole buf to user.
 */
static int __read_socket(socket_info *sinfo, int *len, tsock_ep *ep){

    int res;

    /*    init the len */
    *len = 0;

    switch(sinfo->prtl){

        case SYS_TCP:
redo_tcp:
            res = t_recv(sinfo->fd, sinfo->recv_buf,
                sinfo->recv_len, MSG_NOSIGNAL);

            if(res > 0){
                /* store it into the len */
                *len = res;
                sinfo->num_read++;
            }else if(0 == res){
                /* the link is shut down gracefully */
                return -1;
            }else{
                /* take action per errno */
                switch(errno){

                    case EINTR:
                        LOG(_DBG, "interrupted, try again.");
                        goto redo_tcp;
                        break;

                    case EWOULDBLOCK:
                        /* Data is not ready, return to calling function */
                        LOG(_RTM, "EWOULDBLOCK for socket[%d].", sinfo->fd);
                        return 0;

                    case ETIMEDOUT:
                        LOG(_CRI, "ETIMEDOUT when reading socket[%d].",
                            sinfo->fd);
                        return -1;

                    default:

                        LOG(_CRI, "get a critical error on socket[%d], "
                            "errno[%d](%s).",
                            sinfo->fd, errno, strerror(errno));
                        return -1;
                }
            }
            break;

        case SYS_UDP:

            res = t_recvfm(sinfo->fd, sinfo->recv_buf,
                sinfo->recv_len, 0, ep);

            if(res < 0){
                LOG(_CRI, "recv_from UDP failed with "
                    "socket %d, errno %d(%s).",
                    sinfo->fd, errno, strerror(errno));
                return -1;
            }

            sinfo->num_read++;
            *len = res;

            break;
        case SYS_SCTP:
            /* TODO SCTP */
            return -1;
        default:
            LOG(_CRI, "unsupported protocol %d for socket %d.",
                sinfo->prtl, sinfo->fd);
            return -1;
    }

    return res;
}

static int __trans_send(socket_info *info, char* buf, int len, tsock_ep* ep){

    int res;

    register char* begin = buf;
    register int remain_len = len;
    register int retry_times = 3;

    char buff[INET6_ADDRSTRLEN_BUF];

    do{
        if(NULL == ep){
            res = t_send(info->fd, begin, remain_len, MSG_NOSIGNAL);
        }else{
            res = t_sendto(info->fd, begin, remain_len, MSG_NOSIGNAL, ep);
        }

        if(res <= 0){

            if(EINTR == errno){
                continue;
            }
            /*
             * The difference between EINTR and EAGAIN is
             * send could be blocked in the send forever
             */
            if(EWOULDBLOCK == errno || EAGAIN == errno){

                if(retry_times){
                    --retry_times;
                    __nanosleep(10);
                    continue;
                }
                /* else, error case */
            }

            CHKLOGLVL(L_CRI){

                buff[INET6_ADDRSTRLEN] = '\0';
                LOG(_CRI, "send to remote peer(%s) failed, port[%d]"
                    "errno[%d](%s) for socket fd[%d], buflen[%d].",
                    ep_to_addr(ep, buff, INET6_ADDRSTRLEN_BUF),
                    ep->port, errno, strerror(errno),
                    info->fd, len);

                return -1;
            }
        }

        /* Send sueccfully, buf move len deduce, reset the retry times */
        begin += res;
        remain_len -= res;
        retry_times = 3;

    }while(remain_len > 0); /* remain_len == 0 may cause an infinate loop */

    return len - remain_len;
}

static int __trans_close(socket_info *info){

    register thread_info *trd = info->thread;
    char buf[INET6_ADDRSTRLEN_BUF];
    buf[INET6_ADDRSTRLEN] = '\0';


    if(S_IDEL == info->state){
        return TRANS_SUECCESS;
    }

    /*
     * No matter what kind of status now it is holding
     *     SOCK_CONN_PENDING
     *    SOCK_LISTEN
     *    SOCK_CONNECT
     *    SOCK_CLOSE_PENDING
     *
     *    it will remove the releated socket and decrease the thread active
     *    number of sockets, and reset the socket slot.
     */

    if(is_multicast(&info->r_ep)){

        /* a cross check here */
        if(!(info->opts.opts & SOCK_OPT_LOCAL_PORT)){
            LOG(_ASS, "encouter a bad entry at thread[%d],"
                "there's no local port but with a multicast remote"
                "[%s] for sock fd[%d].",
                (int)(trd - m_info),
                inet_ntop(AF_INET6, &info->r_ep.addr.addr_6, buf,
                    INET6_ADDRSTRLEN),
                info->fd);
            return TRANS_FAILED;
        }

        if(TRANS_SUECCESS != _unset_multicast(info->fd, &info->l_ep,
            &info->r_ep)){

            LOG(_CRI, "fail to leave the multicast group.");

            return TRANS_FAILED;
        }
    }

    /* issue 1, continues here, clean the S_LISTEN_DONE state */
    if((S_LISTEN_DONE == info->state)
        && (info->opts.opts & SOCK_OPT_LISTEN_ONCE)){

        LOG(_RTM, "close the listen once and done, reset");

        /* a double check here to conform it is done */
        if(-1 != info->fd){
            ev_io_stop(trd->base_loop, &info->ev);
            t_close(info->fd);
            trd->active_sock--;
        }

        _reset_socket_info(info, 0);
        return TRANS_SUECCESS;
    }

    /* Kill the internal timer if have */
    if(info->t_llc){
        (void)__kill_sock_timer(info);
    }

    /* A defensive check here */
    if(-1 != info->fd){
        ev_io_stop(trd->base_loop, &info->ev);
        t_close(info->fd);
        trd->active_sock--;
    }
    _reset_socket_info(info, 0);

    return TRANS_SUECCESS;
}

const char* trans_strerror(int _trans_err){
    switch(_trans_err){
        case TRANS_FAILED:
            return "Operation Failed";
        case TRANS_ALREADY_INITED:
            return "Already initialized";
        case TRANS_NOT_INITIALIZE:
            return "Not Initialzed Yet";
        case TRANS_NOT_ENOUGH_MEM:
            return "Not Enough Memory";
        case TRANS_INVALID_VALUE:
            return "Invalid Value";
        case TRANS_A_NULL_POINTER:
            return "Encourter NULL Pointer";
        case TRANS_UNSUPPORTED:
            return "Doesn't Support Yet";
        case TRANS_CROSSCHK_FAIL:
            return "Cross Check Fail";
        case TRANS_EXCEED_LIMITS:
            return "Exceed Limits";
        case TRANS_NOT_FOUND:
            return "No Avaliable Resource";
        case TRANS_LIBEV_ERROR:
            return "Libev Gives An Error";
        case TRANS_TIME_OUT:
            return "Time Out";
        case TRANS_INVALID_TYPE:
            return "Type Is Invalid";
        default:
            return "Unknow Error";
    }
}

WEAK char* trans_version(void){
    return verbuf;
}
